package report


import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.Jedis
import utils.{CleanDataUtils, ReditsUtils, JedisPools}

/**
  * Created by 王康 on 2018/6/30.
  */
object Media_analysis_Redis {

  def main(args: Array[String]) {
    val conf: SparkConf = new SparkConf().setAppName("Media_analysis_Redis").setMaster("local[4]")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ssc: SparkContext = new SparkContext(conf)
    val context: SQLContext = new SQLContext(ssc)

    //整理出AppName集合
    val file: RDD[String] = ssc.textFile("E:\\大数据资料\\project2\\data\\app_dict.txt")
    //将AppName_dict 写入redits 数据库中
    /*ReditsUtils.putApp_dict2Redis(file)*/


    val parquet: DataFrame = context.read.parquet("E:\\大数据资料\\project2\\parquet_1.6.3test")
    val rdd: RDD[Row] = parquet.rdd
    val resultdata: RDD[(String, (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = rdd.mapPartitions(it => {
      val jedis1: Jedis = JedisPools.getJedis()

      val map: Iterator[(String, (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = it.map(row => {
        val appid: String = row.getString(13)
        var appname: String = row.getString(14)

        if (StringUtils.isEmpty(appname)) {
          val hget: String = jedis1.hget("Media_analysis_Redis", appid)
          if (StringUtils.isEmpty(hget)) {
            appname = appid
          } else {
            appname = hget
          }
        }
        val (winprice: Double, adpayment: Double, flag1: Boolean,
        flag2: Boolean, flag3: Boolean, flag4: Boolean, flag5: Boolean,
        flag6: Boolean, flag7: Boolean) = CleanDataUtils.cleanDataByAppNameCore(row)

        (appname, (if (flag1) 1 else 0, if (flag2) 1 else 0, if (flag3) 1 else 0, if (flag4) 1 else 0, if (flag5) 1 else 0, if (flag6) 1 else 0, if (flag7) 1 else 0, if (flag5) winprice else 0.0, if (flag5) adpayment else 0.0))
      })
      jedis1.close()
      map
    })
    val result: RDD[(String, (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = resultdata.reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2, x._3 + y._3, x._4 + y._4, x._5 + y._5, x._6 + y._6, x._7 + y._7, x._8 + y._8, x._9 + y._9))
    result.saveAsTextFile("E:\\大数据资料\\project2\\result\\Media_analysis_Redis")
    result.foreach(t => println(t))
  }

}